// package com.happy.qademo.websheet.config;


// import com.alibaba.fastjson.JSON;
// import com.alibaba.fastjson.JSONObject;
// import com.happy.qademo.websheet.common.ResponseDTO;
// import com.happy.qademo.websheet.service.IMessageProcess;
// import com.happy.qademo.websheet.utils.PakoGzipUtils;
// import lombok.extern.slf4j.Slf4j;
// import org.springframework.beans.factory.annotation.Autowired;
// import org.springframework.stereotype.Component;
// import org.springframework.util.StringUtils;

// import javax.annotation.PostConstruct;
// import javax.websocket.*;
// import javax.websocket.server.PathParam;
// import javax.websocket.server.ServerEndpoint;
// import java.io.IOException;
// import java.util.HashMap;
// import java.util.Map;
// import java.util.concurrent.ConcurrentHashMap;


// @Slf4j
// @ServerEndpoint("/ws/{userId}/{gridKey}")
// @Component
// public class WebSocketServer {
//     private static WebSocketServer webSocketServer;
//     /**
//      * 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
//      */
//     private static int onlineCount = 0;
//     /**
//      * concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
//      */
//     private static ConcurrentHashMap<String, Map<String, WebSocketServer>> webSocketMap = new ConcurrentHashMap<>();
//     /**
//      * 与某个客户端的连接会话，需要通过它来给客户端发送数据
//      */
//     private Session session;
//     /**
//      * 接收userId
//      */
//     private String userId = "";
//     /**
//      * 表格主键
//      */
//     private String gridKey = "";

//     // private String sheetKey = "";

//     @Autowired
//     private IMessageProcess messageProcess;


//     @PostConstruct //通过@PostConstruct实现初始化bean之前进行的操作
//     public void init() {
//         webSocketServer = this;
//         webSocketServer.messageProcess = this.messageProcess;
//     }

//     /**
//      * 连接建立成功调用的方法
//      */
//     @OnOpen
//     public void onOpen(Session session, @PathParam("userId") String userId, @PathParam("gridKey") String gridKey) {
//         // 设置最大消息大小，这里设置为 10MB，你可以根据需要调整
//         session.setMaxTextMessageBufferSize(20 * 1024 * 1024);
//         session.setMaxBinaryMessageBufferSize(20 * 1024 * 1024);

//         this.session = session;
//         this.userId = userId;
//         this.gridKey = gridKey;
//         if (webSocketMap.containsKey(gridKey)) {
//             webSocketMap.get(gridKey).put(userId, this);
//         } else {
//             Map<String, WebSocketServer> map = new HashMap<>();
//             map.put(userId, this);
//             webSocketMap.put(gridKey, map);
//         }
//         addOnlineCount();


//         log.info("用户连接:" + userId + ",打开的表格为：" + gridKey + ",当前在线人数为:" + getOnlineCount());


//     }

//     /**
//      * 连接关闭调用的方法
//      */
//     @OnClose
//     public void onClose() {
//         if (webSocketMap.containsKey(this.gridKey)) {
//             webSocketMap.get(this.gridKey).remove(this.userId);
//             if (webSocketMap.get(this.gridKey).isEmpty()) {
//                 webSocketMap.remove(this.gridKey);
//             }
//         }
//         subOnlineCount();
//         log.info("用户退出:" + this.userId + ",打开的表格为：" + this.gridKey + ",当前在线人数为:" + getOnlineCount());
//     }

//     /**
//      * 收到客户端消息后调用的方法
//      *
//      * @param message 客户端发送过来的消息
//      */
//     @OnMessage
//     public void onMessage(String message, Session session) {
//         //可以群发消息
//         //消息保存到数据库、redis
//         if (StringUtils.hasText(message)) {
//             try {
//                 if ("rub".equals(message)) {
//                     return;
//                 }
//                 String unMessage;
//                 try {
//                     log.info("消息解压前报文：" + message);
//                     unMessage = PakoGzipUtils.unCompressURI(message);  
//                     log.info("用户消息:" + userId + ",消息解压后报文:" + unMessage);
//                 } catch (Exception e) {
//                     log.error("消息解压失败，用户ID: " + userId + ", 消息: " + message, e);
//                     return;
//                 }
//                 JSONObject jsonObject = JSON.parseObject(unMessage);
//                 if (!"mv".equals(jsonObject.getString("t"))) {
//                     try {
//                         webSocketServer.messageProcess.process(this.gridKey, jsonObject);
//                     } catch (Exception e) {
//                         log.error("消息处理失败，用户ID: " + userId + ", 消息: " + unMessage, e);
//                     }
//                 }

//                 Map<String, WebSocketServer> sessionMap = webSocketMap.get(this.gridKey);
//                 if (StringUtils.hasText(unMessage)) {
//                     sessionMap.forEach((key, value) -> {

//                         //广播到除了发送者外的其它连接端
//                         if (!key.equals(this.userId)) {
//                             try {
//                                 //如果是mv,代表发送者的表格位置信息
//                                 if ("mv".equals(jsonObject.getString("t"))) {
//                                     value.sendMessage(JSON.toJSONString(ResponseDTO.mv(userId, userId, unMessage)));

//                                     //如果是切换sheet，则不发送信息
//                                 } else if(!"shs".equals(jsonObject.getString("t"))) {
//                                     value.sendMessage(JSON.toJSONString(ResponseDTO.update(userId, userId, unMessage)));
//                                 }
//                             } catch (Exception e) {
//                                 e.printStackTrace();
//                                 log.error("消息广播失败，目标用户ID: " + key + ", 消息: " + unMessage, e);
//                             }
//                         }
//                     });
//                 }
//             } catch (Exception e) {
//                 log.error("消息处理异常，用户ID: " + userId + ", 消息: " + message, e);
//                 e.printStackTrace();
//             }
//         }
//     }

//     /**
//      * @param session
//      * @param error
//      */
//     @OnError
//     public void onError(Session session, Throwable error) {
//         log.error("用户错误:" + this.userId + ",原因:" + error.getMessage());
//         error.printStackTrace();
//     }

//     /**
//      * 实现服务器主动推送
//      */
//     public void sendMessage(String message) throws IOException, EncodeException {
// //        this.session.getAsyncRemote().sendText(message);
//         synchronized (this.session){
//             this.session.getBasicRemote().sendText(message);
//         }
//     }


//     public static synchronized int getOnlineCount() {
//         return onlineCount;
//     }

//     public static synchronized void addOnlineCount() {
//         WebSocketServer.onlineCount++;
//     }

//     public static synchronized void subOnlineCount() {
//         WebSocketServer.onlineCount--;
//     }
// }








package com.happy.qademo.websheet.config;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.happy.qademo.websheet.common.ResponseDTO;
import com.happy.qademo.websheet.service.IMessageProcess;
import com.happy.qademo.websheet.utils.PakoGzipUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.*;

@Slf4j
@ServerEndpoint("/ws/{userId}/{gridKey}")
@Component
public class WebSocketServer {
    private static WebSocketServer webSocketServer;
    /**
     * 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
     */
    private static int onlineCount = 0;
    /**
     * concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
     */
    private static ConcurrentHashMap<String, Map<String, WebSocketServer>> webSocketMap = new ConcurrentHashMap<>();
    /**
     * 与某个客户端的连接会话，需要通过它来给客户端发送数据
     */
    private Session session;
    /**
     * 接收userId
     */
    private String userId = "";
    /**
     * 表格主键
     */
    private String gridKey = "";

    // 过期时间，单位：秒
    private static final int SESSION_EXPIRE_TIME = 7200;

    // 定时任务调度器
    private static final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(10);

    // 存储每个连接的定时任务
    private ScheduledFuture<?> expireTask;

    @Autowired
    private IMessageProcess messageProcess;

    @PostConstruct //通过@PostConstruct实现初始化bean之前进行的操作
    public void init() {
        webSocketServer = this;
        webSocketServer.messageProcess = this.messageProcess;
    }

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId, @PathParam("gridKey") String gridKey) {
        // 设置最大消息大小
        session.setMaxTextMessageBufferSize(80 * 1024 * 1024);
        session.setMaxBinaryMessageBufferSize(80 * 1024 * 1024);

        this.session = session;
        this.userId = userId;
        this.gridKey = gridKey;
        if (webSocketMap.containsKey(gridKey)) {
            webSocketMap.get(gridKey).put(userId, this);
        } else {
            Map<String, WebSocketServer> map = new HashMap<>();
            map.put(userId, this);
            webSocketMap.put(gridKey, map);
        }
        addOnlineCount();

        log.info("用户连接:" + userId + ",打开的表格为：" + gridKey + ",当前在线人数为:" + getOnlineCount());

        // 启动过期定时任务
        startExpireTask();
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        if (webSocketMap.containsKey(this.gridKey)) {
            webSocketMap.get(this.gridKey).remove(this.userId);
            if (webSocketMap.get(this.gridKey).isEmpty()) {
                webSocketMap.remove(this.gridKey);
            }
        }
        subOnlineCount();
        log.info("用户退出:" + this.userId + ",打开的表格为：" + this.gridKey + ",当前在线人数为:" + getOnlineCount());

        // 取消过期定时任务
        cancelExpireTask();
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        // 每次收到消息，重置过期时间
        resetExpireTask();
        if (userId == null || userId.isEmpty()) {
            try {
                if (session.isOpen()) {
                    session.close();
                }
            } catch (IOException e) {
                log.error("关闭会话时出错: " , e);
            }
        }

        //可以群发消息
        //消息保存到数据库、redis
        if (StringUtils.hasText(message)) {
            try {
                if ("rub".equals(message)) {
                    return;
                }
                String unMessage;
                unMessage = PakoGzipUtils.unCompressURI(message);
                log.info("用户:" + userId + " ,消息解压后报文:" + unMessage);
                JSONObject jsonObject = JSON.parseObject(unMessage);
                if (!"mv".equals(jsonObject.getString("t"))) {
                   webSocketServer.messageProcess.process(this.gridKey, jsonObject);
                   
                }

                Map<String, WebSocketServer> sessionMap = webSocketMap.get(this.gridKey);
                if (StringUtils.hasText(unMessage)) {
                    sessionMap.forEach((key, value) -> {
                        //广播到除了发送者外的其它连接端
                        if (!key.equals(this.userId)) {
                            try {
                                //如果是mv,代表发送者的表格位置信息
                                if ("mv".equals(jsonObject.getString("t"))) {
                                    value.sendMessage(JSON.toJSONString(ResponseDTO.mv(userId, userId, unMessage)));

                                    //如果是切换sheet，则不发送信息
                                } else if (!"shs".equals(jsonObject.getString("t"))) {
                                    value.sendMessage(JSON.toJSONString(ResponseDTO.update(userId, userId, unMessage)));
                                }
                           } catch (Exception e) {
                                e.printStackTrace();
                                 log.error("消息广播失败，目标用户ID: " + key + ", 消息: " + unMessage, e);
                            }
                        }
                    });
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("用户错误:" + this.userId + ",原因:" + error.getMessage());
        error.printStackTrace();

        // 取消过期定时任务
        cancelExpireTask();
    }

    /**
     * 实现服务器主动推送
     */
    public void sendMessage(String message) throws IOException, EncodeException {
//        this.session.getAsyncRemote().sendText(message);
        synchronized (this.session) {
            this.session.getBasicRemote().sendText(message);
        }
    }

    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    public static synchronized void addOnlineCount() {
        WebSocketServer.onlineCount++;
    }

    public static synchronized void subOnlineCount() {
        WebSocketServer.onlineCount--;
    }

    /**
     * 启动过期定时任务
     */
    private void startExpireTask() {
        expireTask = scheduler.schedule(() -> {
            try {
                log.info("用户 " + userId + " 会话过期，关闭连接");
                if (session.isOpen()) {
                    session.close();
                }
            } catch (IOException e) {
                log.error("关闭会话时出错，用户ID: " + userId, e);
            }
        }, SESSION_EXPIRE_TIME, TimeUnit.SECONDS);
    }

    /**
     * 取消过期定时任务
     */
    private void cancelExpireTask() {
        if (expireTask != null) {
            expireTask.cancel(false);
        }
    }

    /**
     * 重置过期定时任务
     */
    private void resetExpireTask() {
        cancelExpireTask();
        startExpireTask();
    }
}